package com.weige.elefence.mq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.weige.elefence.utils.ReadJsonFileUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * 生产者生产消息
 * https://blog.csdn.net/weixin_43891622/article/details/115660444
 *
 * @author ltj
 */
public class Product {
    private static final Logger log = LoggerFactory.getLogger(Product.class);
    public static Channel channel;
    public static void run() {
        Map<String, String> map = ReadJsonFileUtil.map;
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 连接IP
        String host = map.get("host");
        connectionFactory.setHost(host);
        String port = map.get("port");
        // 连接端口
        connectionFactory.setPort(Integer.parseInt(port));
        // 虚拟主机(默认为/)
        connectionFactory.setVirtualHost(map.get("virtual-host"));
        // 账号 默认是guest
        connectionFactory.setUsername(map.get("username"));
        // 密码 默认是guest
        connectionFactory.setPassword(map.get("password"));
        try {
            // 创建连接
            Connection connection = connectionFactory.newConnection();
            log.info("mq = " + host + ":" + port + "   连接成功！");
            // 创建信道
            channel = connection.createChannel();

            // 生产者发送消息(注意点:不要提前关闭信道与连接因为失败通知可能还没回来,找不到交换机不会触发失败通知,只有在找不到队列的时候才触发失败通知)
            // 第一个参数:交换机名字,第二个参数:路由键,第三个参数:是否开启失败通知,第四个参数为:不支持的参数,第五个参数为:消息的头帧,第六个参数为:消息体
            channel.addReturnListener((replycode, replyText, exchange, routingKey, basicProperties, bytes) -> {
                System.out.println("交换机为:" + exchange);
                System.out.println("路由键为:" + routingKey);
                System.out.println("消息体为:" + new String(bytes));
                System.out.println("消息头帧为:" + basicProperties);
                System.out.println("错误文本为:" + replyText);
                System.out.println("错误码为:" + replycode);
            });

            // 在该信道上开启发送者确认
            channel.confirmSelect();

            byte[] bytes = "msg".getBytes();
            Map<String, Object> headers = new HashMap<>(5);
            headers.put("latitude",  51.5252949);
            headers.put("longitude", -0.0905493);
            AMQP.BasicProperties build = new AMQP.BasicProperties.Builder()
                    // 头部
                    .headers(headers)
                    //有效期
                    .expiration("60000")
                    .build();
            // 第一个是交换机名称，第二个是队列的路由键
            channel.basicPublish("exchangeEnv", "env.userAccess_Nari.#", build,bytes );

            boolean b = channel.waitForConfirms();
            if (b) {
                log.info("发送成功");
            } else {
                log.info("发送失败");
            }
        } catch (Exception e) {
            log.info("MQ Connection timed out: connect");
//            e.printStackTrace();
        }

    }
}